package com.lightboat.system.queue;

import cn.hutool.core.util.ObjUtil;
import cn.hutool.extra.spring.SpringUtil;
import org.redisson.RedissonShutdownException;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RFuture;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;


/**
 * @Description: 生产者消费者延时队列
 * @Package: com.mz.cache
 * @ClassName: DelaySaver
 * @Author: 小政同学  Mail：it_xiaozheng@163.com
 * @date: 2023/1/14/014 9:14
 */
public class DelaySaver<T> {

    private static final ExecutorService fixedThreadPool = Executors.newFixedThreadPool(200);
    Logger log = LoggerFactory.getLogger(DelaySaver.class);
    private RBlockingQueue<T> queue = null;
    private RDelayedQueue<T> delayedQueue = null;
    private Consumer<T> comer;
    private String threadName;
    private int num;
    private final Thread consumerThread = new Thread(() -> {
        for (int i = 0; i < num; i++) {
            int finalI = i;
            log.info("启动线程:{}", threadName + finalI);
            fixedThreadPool.execute(() -> {
                // 指定线程名称
                Thread.currentThread().setName(threadName.concat("_") + finalI);
                while (true) {
                    try {
                        comer.consume(queue.take());
                    } catch (Exception e) {
                        if (e instanceof RedissonShutdownException) {
                            log.error("redisson延时队列：{}", e.getMessage());
                            break;
                        } else {
                            log.error("redisson延时队列：{}-》{}", threadName.concat("_") + finalI, e.getMessage());
                        }
                    }
                }
            });
        }
    }, "DelaySaver-consumer");

    private DelaySaver() {
    }

    public static <T> DelaySaver<T> create(RedissonClient redisson, String key, String threadName, int num) {
        Assert.isTrue(num > 0, "启动线程数需大于0");
        Assert.isTrue(ObjUtil.isNotEmpty(key), "订阅队列不能为空");
        Assert.isTrue(ObjUtil.isNotEmpty(threadName), "线程名称不能为空");
        DelaySaver de = new DelaySaver();
        RBlockingQueue<T> blockingQueue = redisson.getBlockingQueue(key + "_" + SpringUtil.getActiveProfile());
        de.queue = blockingQueue;
        de.threadName = threadName;
        de.num = num;
        de.delayedQueue = redisson.getDelayedQueue(blockingQueue);
        return de;
    }

    public void updateComsumer(Consumer<T> comer) {
        this.comer = comer;
    }

    public void start() {
        if (null == comer) {
            log.error("消费者不存在，请先配置消费者后启动!");
            return;
        }
        log.info("启动消费者线程:{}", threadName);
        consumerThread.start();
    }

    public boolean publishAsync(T task, long delay, TimeUnit timeUnit) {
        if (null == comer) {
            log.error("消费者不存在，请先配置消费者后启动!");
            return false;
        }
        return delayed(task, delay, timeUnit);
    }

    /**
     * 默认延时单位为秒
     *
     * @param task
     * @param delay
     * @return
     */
    public boolean publishAsync(T task, long delay) {
        if (null == comer) {
            log.error("消费者不存在，请先配置消费者后启动!");
            return false;
        }
        return delayed(task, delay, null);
    }

    private boolean delayed(T task, long delay, TimeUnit timeUnit) {
        RFuture<Void> future = delayedQueue.offerAsync(task, delay, ObjUtil.isEmpty(timeUnit) ? TimeUnit.SECONDS : timeUnit);
        if (future.toCompletableFuture().isDone()) {
            throw new SecurityException("生产消费队列异常：" + threadName);
        }
        log.info("添加阻塞队列数据：{}", task);
        return true;
    }

    /**
     * 默认延时单位为秒
     *
     * @param task
     * @return
     */
    public boolean delAsync(T task) {
        boolean remove = queue.remove(task);
        boolean remove1 = delayedQueue.remove(task);
        return remove1 || remove;
    }

    public void shutdown() {
        fixedThreadPool.shutdown();
    }

    public interface Consumer<T> {

        void consume(T task);
    }
}
